package com.itc.bi.service.impl;

import com.alibaba.fastjson2.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.itc.bi.constant.KafkaConstant;
import com.itc.bi.constant.RedisConstant;
import com.itc.bi.constant.SystemConstant;
import com.itc.bi.entity.DataProcessEntity;
import com.itc.bi.entity.DataProcessInfoEntity;
import com.itc.bi.entity.DataSetColumnEntity;
import com.itc.bi.mapper.DataProcessInfoMapper;
import com.itc.bi.mapper.DataProcessMapper;
import com.itc.bi.mapper.DataSetColumnMapper;
import com.itc.bi.query.DataSetColumnQuery;
import com.itc.bi.service.DataSetColumnService;
import com.itc.bi.vo.excelVo.HeardVO;
import com.itc.common.core.exception.CheckedException;
import com.itc.common.core.exception.ServiceException;
import org.apache.commons.lang3.StringUtils;
import org.redisson.api.RList;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.*;
import java.util.concurrent.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
* @author xuzy
*/
@Service
public class DataSetColumnServiceImpl extends ServiceImpl<DataSetColumnMapper, DataSetColumnEntity>
    implements DataSetColumnService {
    @Autowired
    KafkaTemplate kafkaTemplate;
    @Autowired
    RedissonClient redissonClient;
    @Autowired
    DataProcessInfoMapper dataProcessInfoMapper;
    @Autowired
    DataProcessMapper dataProcessMapper;

    @Override
    public List<DataSetColumnEntity> list(DataSetColumnQuery query){
        if(null == query.getDataSetId()){
            return new ArrayList<>();
        }
        if(null != query.getProcessInfoId()){
            DataProcessInfoEntity info = dataProcessInfoMapper.selectById(query.getProcessInfoId());
            if(null == info){
                return new ArrayList<>();
            }
            LambdaQueryWrapper<DataSetColumnEntity> qw = Wrappers.lambdaQuery();
            qw.eq(DataSetColumnEntity::getDataSetId,info.getDpId());
            qw.lt(DataSetColumnEntity::getStepNum,info.getOrderNum());
            List<DataSetColumnEntity> list = super.list(qw);
            list.stream().forEach(item->{
                item.setType(item.getColumnType());
            });
            return list;
        }

        return listById(query.getDataSetId());
    }

    /**
     * 修改数据集字段的类型和注释
     * @param entity
     * @return
     */
    @Override
    public boolean updateById(DataSetColumnEntity entity){
        if(null == entity.getId()){
            throw new CheckedException("ID不存在");
        }
        DataSetColumnEntity old = getById(entity.getId());
        if(null == old){
            throw new CheckedException("记录不存在");
        }
        if(StringUtils.isNoneBlank(entity.getColumnComment())){
            old.setColumnComment(entity.getColumnComment());
        }
        old.setColumnType(entity.getColumnType());
        updateById(old);
        return true;
    }

    /**
     * 批量修改数据集字段信息,只能是同一数据集
     * @param list
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public void batch(List<DataSetColumnEntity> list){
        if(null == list || list.size() < 1){
            return;
        }
        Set<Long> ids = list.stream().filter(item -> null != item.getId()).map(item -> item.getId()).collect(Collectors.toSet());
        if(list.size() != ids.size()){
            throw new ServiceException("ID不能为空或重复,请检查");
        }
        LambdaQueryWrapper<DataSetColumnEntity> qw = Wrappers.lambdaQuery();
        qw.in(DataSetColumnEntity::getId,ids);
        List<DataSetColumnEntity> dataSetList = list(qw);
        if(list.size() != dataSetList.size()){
            Set<Long> oldIds = dataSetList.stream().map(item -> item.getId()).collect(Collectors.toSet());
            ids.removeAll(oldIds);
            throw new ServiceException(ids.toString()+"不存在");
        }
        Set<Long> datasetId = dataSetList.stream().map(item -> item.getDataSetId()).collect(Collectors.toSet());
        if(datasetId.size() > 1){
            throw new ServiceException("不是同一数据集数据,不能操作");
        }
        Long dtId = dataSetList.get(0).getDataSetId();

        Map<Long,DataSetColumnEntity> columnMap = dataSetList.stream().collect(Collectors.toMap(DataSetColumnEntity::getId, Function.identity()));

        boolean typeChangeFlag = false;

        for (DataSetColumnEntity column: list){
            DataSetColumnEntity oldColumn = columnMap.get(column.getId());

            if(!StringUtils.equalsAny(column.getColumnType(),SystemConstant.DATA_TYPE_NUMBER,SystemConstant.DATA_TYPE_DATE,SystemConstant.DATA_TYPE_STRING)){
                throw new ServiceException("类型"+column.getColumnType()+"暂不支持");
            }
            if(!oldColumn.getColumnComment().equals(column.getColumnComment()) || !oldColumn.getColumnType().equals(column.getColumnType())){
                String type = column.getColumnType();
                String oldType = oldColumn.getColumnType();
                if(!type.equals(oldType)){
                    typeChangeFlag = true;
                    if(!StringUtils.equalsAny(SystemConstant.DATA_TYPE_STRING,type,oldType)){
                        throw new ServiceException("字段类型中日期和数字之间不能想换转换");
                    }
                }
                oldColumn.setColumnComment(column.getColumnComment());
                oldColumn.setColumnType(column.getColumnType());
            }else{//没有更新的数据剔除掉不更新数据库
                columnMap.remove(column.getId());
            }
        }
        if(columnMap.size()>0){
            //更新
            updateBatchById(columnMap.values());
            //根据ID反查,字段和类对应
            qw.clear();
            qw.in(DataSetColumnEntity::getId,columnMap.values().stream().map(item->item.getId()).collect(Collectors.toList()));
            Map<String, DataSetColumnEntity> map = list(qw).stream().collect(Collectors.toMap(DataSetColumnEntity::getColumnName, Function.identity()));

            //数据集对应的加工id
            List<Long> processIds = dataProcessMapper.listIdByDtId(dtId);
            if(processIds.size() > 0){
                //查询需要更新的数据
                qw.clear();
                qw.in(DataSetColumnEntity::getDataSetId,processIds);
                qw.in(DataSetColumnEntity::getColumnName,map.keySet());
                qw.eq(DataSetColumnEntity::getStepNum,0);
                List<DataSetColumnEntity> updateList = list(qw);
                updateList.stream().forEach(item->{
                    item.setColumnType(map.get(item.getColumnName()).getColumnType());
                    item.setColumnComment(map.get(item.getColumnName()).getColumnComment());
                });
                if(updateList.size()>0){
                    updateBatchById(updateList);
                }
                //更新缓存
                processIds.stream().forEach(id->{
                    redisCache(id);
                });
            }
            redisCache(dtId);
            //类型有变化时需要特殊处理
            if(typeChangeFlag){
                JSONObject data = new JSONObject();
                data.put("dtId",dtId);
                kafkaTemplate.send(KafkaConstant.KAFKA_TOPIC_DATA_SET_COLUMN_EDIT,data.toJSONString());
            }
        }
    }

    @Override
    public void insertList(List<DataSetColumnEntity> columnList){
        if(CollectionUtils.isEmpty(columnList)){
            return;
        }
        Integer num = super.baseMapper.getPositionByDtId(columnList.get(0).getDataSetId());
        if(null == num){
            num = 0;
        }
        for (DataSetColumnEntity column : columnList){
            num++;
            column.setColumnPosition(num);
        }
        saveBatch(columnList);
    }

    @Override
    public void redisCache(Long dataSetId){
        RList<Object> redisList = redissonClient.getList(RedisConstant.DATA_SET_COLUMN + dataSetId);
        redisList.clear();

        listById(dataSetId);
    }

    @Override
    public List<DataSetColumnEntity> columnListById(Long dataSetId){
        return listById(dataSetId);
    }

    @Override
    public List<HeardVO> heardListById(Long dataSetId){
        List<HeardVO> list = new ArrayList<>();
        List<DataSetColumnEntity> columnList = listById(dataSetId);
        for (DataSetColumnEntity column : columnList){
            list.add(new HeardVO(column.getColumnName(),column.getColumnComment(),column.getColumnType()));
        }
        return list;
    }

    private List<DataSetColumnEntity> listById(Long dataSetId){
        RList<DataSetColumnEntity> redisList = redissonClient.getList(RedisConstant.DATA_SET_COLUMN + dataSetId);
        List<DataSetColumnEntity> list = new ArrayList<>();
        if(redisList.isExists()){
            list.addAll(redisList.readAll());
            redisList.expire(2, TimeUnit.DAYS);
            if(list.size() > 0){
                return  list;
            }
        }
        RLock lock = redissonClient.getLock("datasetcolumninit:"+dataSetId);
        lock.lock();
        try {
            if(redisList.isExists()){
                list.addAll(redisList.readAll());
                if(list.size() > 0){
                    return  list;
                }
            }
            LambdaQueryWrapper<DataSetColumnEntity> qw = Wrappers.lambdaQuery();
            qw.eq(DataSetColumnEntity::getDataSetId,dataSetId);
            qw.orderBy(true,true,DataSetColumnEntity::getColumnPosition);
            List<DataSetColumnEntity> columnList = super.list(qw);
            if(columnList.size()>0){
                columnList.stream().forEach(item-> item.setType(item.getColumnType()));
                redisList.clear();
                redisList.addAll(columnList);
                redisList.expire(2, TimeUnit.DAYS);
            }
            list.addAll(columnList);

            return list;
        }finally {
            if(lock != null){
                lock.unlock();
            }
        }
    }

    @Override
    public void validColumn(Long dtId,List<DataSetColumnEntity> columns){
        //加工数据
        List<DataProcessEntity> processList = dataProcessMapper.selectList(Wrappers.lambdaQuery(DataProcessEntity.class).eq(DataProcessEntity::getDtId, dtId));
        if(CollectionUtils.isEmpty(processList)){
            return;
        }

        Map<String, String> columnToTypeMap = columns.stream().collect(Collectors.toMap(DataSetColumnEntity::getColumnName, DataSetColumnEntity::getColumnType));

        //原始字段
        List<DataSetColumnEntity> oldList = super.list(Wrappers.lambdaQuery(DataSetColumnEntity.class).eq(DataSetColumnEntity::getDataSetId, dtId));
        Iterator<DataSetColumnEntity> it = oldList.iterator();
        while (it.hasNext()){
            DataSetColumnEntity item = it.next();
            if(columnToTypeMap.containsKey(item.getColumnName()) && item.getColumnType().equals(columnToTypeMap.get(item.getColumnName()))){
                it.remove();
            }
        }
        //无删除字段直接返回
        if(oldList.size() == 0){
            return;
        }

        //删除的字段
        List<String> deleteColumns = oldList.stream().map(item -> item.getColumnName()).collect(Collectors.toList());

        List<Long> processListIds = processList.stream().map(item -> item.getId()).collect(Collectors.toList());
        List<DataProcessInfoEntity> infoList = dataProcessInfoMapper.selectList(Wrappers.lambdaQuery(DataProcessInfoEntity.class).isNotNull(DataProcessInfoEntity::getUserField).in(DataProcessInfoEntity::getDpId, processListIds));

        List<String> useColumns = new ArrayList<>();
        infoList.stream().filter(item->StringUtils.isNotBlank(item.getUserField())).forEach(item->{
            useColumns.addAll(Arrays.asList(item.getUserField().split("、")));
        });

        //删除并有使用的字段
        List<String> deleteUseColumn = deleteColumns.stream().filter(item -> useColumns.contains(item)).collect(Collectors.toList());
        if(deleteUseColumn.size() > 0){
            throw new ServiceException(String.format("字段 %s 被加工数据使用,不能删除",deleteUseColumn.toString()));
        }
    }
}




